package org.mnj07.jms.activemq;

import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

public class HelloJMS {
  public static void main(String[] args) {
    thread(new Producer(), false);
    thread(new Consumer(), false);
  }

  public static void thread(Runnable runnable, boolean daemon) {
    Thread brokerThread = new Thread(runnable);
    brokerThread.setDaemon(daemon);
    brokerThread.start();
  }

  public static class Producer implements Runnable {
    public void run() {
      try{
        // Create a ConnectionFactory
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("vm://localhost");

        // Create a Connection
        Connection connection = connectionFactory.createConnection();
        connection.start();

        // Create a Session
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        // Create the destination (Topic or Queue)
        Destination destination = session.createQueue("TEST.FOO");

        // Create a MessageProducer from the Session to the Topic or Queue
        MessageProducer producer = session.createProducer(destination);
        producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

        int index = 0;
        
        for(int i=0 ; i<10 ; i++){
          // Create a messages
          String text = "Hello world! Message Index: " + index;
          TextMessage message = session.createTextMessage(text);

          if(i == 9){
            message.setBooleanProperty("EndOfMessages", true);
          }
          
          // Tell the producer to send the message
          System.out.println("Producer: Sent message: [" + text + "]");
          producer.send(message);
          Thread.sleep(1000);
          index++;
        }

        // Clean up
        session.close();
        connection.close();
      }
      catch (Exception e){
        System.out.println("Caught: " + e);
        e.printStackTrace();
      }
    }
  }

  public static class Consumer implements Runnable, ExceptionListener {
    public void run() {
      try{

        // Create a ConnectionFactory
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("vm://localhost");

        // Create a Connection
        Connection connection = connectionFactory.createConnection();
        connection.start();

        connection.setExceptionListener(this);

        // Create a Session
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        // Create the destination (Topic or Queue)
        Destination destination = session.createQueue("TEST.FOO");

        // Create a MessageConsumer from the Session to the Topic or Queue
        MessageConsumer consumer = session.createConsumer(destination);

        boolean shouldRun = true;
        while(shouldRun){
          System.out.println("Consumer: Waiting for a message ...");
          // Wait for a message
          Message message = consumer.receive();

          if(message instanceof TextMessage){
            TextMessage textMessage = (TextMessage) message;
            String text = textMessage.getText();
            System.out.println("Consumer: Received: " + text);
            
            boolean end = textMessage.getBooleanProperty("EndOfMessages");
            if(end){
              System.out.println("Consumer: Got end of messages property");
              shouldRun = false;
            }
          }
          else{
            System.out.println("Consumer: Received: " + message);
          }
        }

        consumer.close();
        session.close();
        connection.close();
      }
      catch (Exception e){
        System.out.println("Caught: " + e);
        e.printStackTrace();
      }
    }

    public synchronized void onException(JMSException ex) {
      System.out.println("JMS Exception occured.  Shutting down client.");
    }
  }
}
